package com.ibm.cps.storm;

import java.util.ArrayList;

import backtype.storm.tuple.Values;

import com.ibm.cps.message.RawDataMessage;
import com.ibm.cps.newmessage.AbstractMetadata;
import com.ibm.cps.processors.DFTProcessorTest;
import com.ibm.cps.processors.ProcessorUtil;
import com.ibm.cps.processors.OutputProcessorTest;
import com.ibm.cps.processors.RuleEngineProcessorTest;
import com.ibm.cps.processors.SchemaTest;
import com.ibm.cps.spark.streaming.TestTuple;
import com.ibm.util.TopologyStreamIds;
import com.ibm.util.exception.CPSException;

public class StormTestUtil {

	public static String tenantid = "123456789012345678901239";
	public static final String schemaid = "0000000008e151b18071eae8";
	public static final String inputProcessorid = "0000000000ed66138a01299f";
	public static final String dftProcessorid = "0000000000e8151c09017eae";
	public static final String naiveRuleProcessorid = "0000000000ceab2e30f31549";
	public static final String ftRuleProcessorid = "0000000000ceab2e30f31441";
	public static final String tfRuleProcessorid = "0000000000ceab2e30f31442";
	public static final String fttfRuleProcessorid = "0000000000ceab2e30f31443";
	public static final String publisherid = "0000000000e8151c09017eaf";

	/**
	 * a topology including a schema, input, rule
	 * 
	 * @return
	 * @throws CPSException
	 */
	public static ArrayList<TestTuple> inputRuleEngineTopology() throws CPSException {
		ArrayList<TestTuple> topology = new ArrayList<TestTuple>();
		String metadataStreamid = TopologyStreamIds.PROCESSOR_SPOUT_STREAM;
		topology.add(constructSchemaTuple(metadataStreamid));
		topology.add(constructInputTuple(metadataStreamid));
		topology.add(constructRuleEngineTuple(ProcessorUtil.inputProcessorid, ProcessorUtil.naiveRuleProcessorid,
				metadataStreamid, "phase0"));
		return topology;
	}

	/**
	 * a topology including a schema, input, rule
	 * 
	 * @return
	 * @throws CPSException
	 */
	public static ArrayList<TestTuple> inputDFTRuleEngineTopology() throws CPSException {
		ArrayList<TestTuple> topology = new ArrayList<TestTuple>();
		String metadataStreamid = TopologyStreamIds.PROCESSOR_SPOUT_STREAM;
		topology.add(constructSchemaTuple(metadataStreamid));
		topology.add(constructInputTuple(metadataStreamid));
		topology.add(constructDFTTuple(metadataStreamid));
		topology.add(constructRuleEngineTuple(ProcessorUtil.dftProcessorid, ProcessorUtil.naiveRuleProcessorid,
				metadataStreamid, "payload(0)"));
		return topology;
	}

	public static TestTuple constructRawTuple(String dataStreamid, int tskey, int timestamp, int... elements)
			throws CPSException {
		RawDataMessage data = ProcessorUtil.createRawData(tskey, timestamp, elements);
		TestTuple rawData = new TestTuple(dataStreamid, new Values(data, data.getTsKey()));
		return rawData;
	}

	public static TestTuple constructSchemaTuple(String metadataStreamid) throws CPSException {
		AbstractMetadata schema = SchemaTest.constructSchemaMetadata(ProcessorUtil.schemaid);
		return new TestTuple(metadataStreamid, new Values(schema));
	}

	public static TestTuple constructInputTuple(String metadataStreamid) throws CPSException {
		// AbstractMetadata inputMetadata = InputProcessorTest
		// .createInputMetadata(ProcessorUtil.schemaid,
		// ProcessorUtil.inputProcessorid);
		AbstractMetadata inputMetadata = null;
		return new TestTuple(metadataStreamid, new Values(inputMetadata));
	}

	public static TestTuple constructDFTTuple(String metadataStreamid) throws CPSException {
		AbstractMetadata dft = DFTProcessorTest.createDFTMetadata(ProcessorUtil.tenantid,
				ProcessorUtil.inputProcessorid, ProcessorUtil.dftProcessorid, 3);
		TestTuple dftTuple = new TestTuple(metadataStreamid, new Values(dft));
		return dftTuple;
	}

	public static TestTuple constructRuleEngineTuple(String parentProcessorid, String ruleProcessorid,
			String metadataStreamid, String ruleAttribute) throws CPSException {
		AbstractMetadata naiveRule = RuleEngineProcessorTest.constructRuleEngineMetadata(parentProcessorid,
				ProcessorUtil.tenantid, 0, ruleProcessorid, ruleAttribute);
		TestTuple ruleTuple = new TestTuple(metadataStreamid, new Values(naiveRule));
		return ruleTuple;
	}

	public static TestTuple constructPublisherTuple(String metadataStreamid) throws CPSException {
		AbstractMetadata publisher = OutputProcessorTest.constructMongoPublisherMetadata();
		TestTuple publisherTuple = new TestTuple(metadataStreamid, new Values(publisher));
		return publisherTuple;
	}
}
